package cn.xuexiyuan.flinkstudy.transformation;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashSet;
import java.util.Set;

/**
 * @Description:
 * @Author 左龙龙
 * @Date 21-3-23
 * @Version 1.0
 **/
public class TransformationDemo01 {

    public static void main(String[] args) throws Exception {
        // 0.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 敏感词
        Set<String> sensitives = new HashSet<String>();
        sensitives.add("TMD");
        sensitives.add("他妈的");

        // 1.source
        DataStreamSource<String> ds1 = env.socketTextStream("localhost", 6666);

        // 2.transformation
        // 过滤敏感词
        SingleOutputStreamOperator<String> filter = ds1.filter(new FilterFunction<String>() {
            @Override
            public boolean filter(String s) throws Exception {
                return sensitives.contains(s) == false;
            }
        });

        SingleOutputStreamOperator<Tuple2<String, Integer>> wordAndOne = filter.flatMap((FlatMapFunction<String, Tuple2<String, Integer>>) (s, collector) -> {
            for (String token : s.split("\\W+")) {
                collector.collect(Tuple2.of(token, 1));
            }
        }).returns(Types.TUPLE(Types.STRING, Types.INT));

        KeyedStream<Tuple2<String, Integer>, String> grouped = wordAndOne.keyBy(d -> d.f0);

        SingleOutputStreamOperator<Tuple2<String, Integer>> result = grouped.reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
                return Tuple2.of(t1.f0, t1.f1 + t2.f1);
            }
        });

        // 3.sink
        result.print();

        // 4.excute
        env.execute("TransformationDemo01");
    }
}
